package com.stone.bigdata.flink;

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.io.jdbc.JDBCOutputFormat;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.types.Row;

import java.sql.*;
import java.util.Arrays;

/**
 * @author fujianjian
 * @date 2023/8/10 11:05
 **/
public class MysqlSinkFunctionV1 {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSource<String[]> datasource = env.fromCollection(Arrays.asList(
                new String[]{"92020744", "92020744", "贺安东", "510724199710120817", "19983549330"},
                new String[]{"92017458", "92017458", "王保同", "341226199608141317", "18721516653"},
                new String[]{"92017079", "92017079", "何政", "421083199208142851", "13995609474"},
                new String[]{"10050751", "10050751", "刘刚", "320722199107285512", "18930692859"},
                new String[]{"92000405", "92000405", "周润发", "360481199305014810","13697085953"}
        ));

        datasource.map(item -> {
            Row row = new Row(5);
            for (int i = 0; i < item.length; i++) {
                row.setField(i, item[i]);
            }
            return row;
        }).output(JDBCOutputFormat.buildJDBCOutputFormat()
                .setDrivername("com.mysql.cj.jdbc.Driver")
                .setDBUrl("jdbc:mysql://10.99.2.101:3306/ihr_core?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true&serverTimezone=GMT%2B8&zeroDateTimeBehavior=convertToNull")
                .setUsername("ihr_master").setPassword("iHR@zhengB!2021")
                .setQuery("insert into fk_emp_temp(id, code, name, idno, phone) values(?,?,?,?,?) on duplicate key update code=values(code),name=values(name), idno=values(idno), phone=values(phone)")
                .setSqlTypes(new int[]{Types.VARCHAR, Types.VARCHAR, Types.VARCHAR, Types.VARCHAR, Types.VARCHAR})
                .finish()).setParallelism(Runtime.getRuntime().availableProcessors());


        env.execute("mysql_jdbc_format_output");
        System.out.println("写入成功!!!!");
    }

    public static class MysqlSinkFunction extends RichSinkFunction<String[]> {

        private Connection connection;
        private PreparedStatement ps;

        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            Class.forName("com.mysql.cj.jdbc.Driver");
            connection = DriverManager.getConnection("jdbc:mysql://10.99.2.101:3306/ihr_core?useUnicode=true&characterEncoding=UTF-8&allowMultiQueries=true&serverTimezone=GMT%2B8&zeroDateTimeBehavior=convertToNull",
                    "ihr_master", "iHR@zhengB!2021");
            Statement stat = connection.createStatement();

        }

        @Override
        public void invoke(String[] value, Context context) throws Exception {
            super.invoke(value, context);
        }

        @Override
        public void close() throws Exception {
            super.close();
        }
    }

}
